package reactiveStream;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

/**
 * @projectName: Webflux_demo
 * @package: reactiveStream
 * @className: ReactiveProcessor 用于转发 Publisher和 subscriber之间消息，并且可以对消息进行处理
 * @author: 冷环渊 doomwatcher
 * @description: TODO
 * @date: 2021/12/14 0:25
 * @version: 1.0
 */
public class ReactiveProcessor extends SubmissionPublisher<String> implements Flow.Processor<String, String> {
    private Flow.Subscription subscription;

    @Override
    public void onSubscribe(Flow.Subscription subscription) {
        this.subscription = subscription;
        System.out.println("Processor建立订阅关系");
        subscription.request(1);
    }

    @Override
    public void onNext(String item) {
        System.out.println("Processor接收数据：" + item);
        //中间处理
        //数据发给最终订阅者
        this.submit(item.toUpperCase());
        //背压的实现核心
        this.subscription.request(1);
    }

    @Override
    public void onError(Throwable throwable) {
        System.out.println("出现错误了");
    }

    @Override
    public void onComplete() {
        System.out.println("数据传输成功");
    }
}
